package com.mxd.flink.connector.table;

import com.mxd.flink.connector.config.Delimiters;
import com.mxd.flink.connector.config.RedisCmd;
import com.mxd.flink.connector.config.RedisOptions;
import com.mxd.flink.connector.operator.RedisOperator;
import com.mxd.flink.connector.operator.RedisOperators;
import com.mxd.flink.connector.utils.RedisRowUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.CollectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * @author rongdi
 * @date 2022/9/24 20:36
 */
public class RedisSinkFunction extends RichSinkFunction {

    private static final Logger LOG = LoggerFactory.getLogger(RedisSinkFunction.class);

    private ReadableConfig config;

    private RedisOperator redisOperator;

    /**
     * 这里是一个可序列化的schema工具对象，主要作用是下面用来把输入的GenericRowData
     * 序列化成指定的format格式，比如format=”json“，那就是把RowData转成json字符串
     */
    private SerializationSchema<RowData> serializationSchema;

    private String key;

    private String command;

    /**
     * 主键列
     */
    private List<String> primaryKeys;

    /**
     * 各字段信息
     */
    private DataType dataType;

    /**
     * 这里需要注意的是一定不要把TableSchema对象放成成员变量，因为他不可序列化，会导致报错
     * @param config
     * @param serializationSchema
     */
    public RedisSinkFunction(ReadableConfig config, SerializationSchema<RowData> serializationSchema, DataType dataType, List<String> primaryKeys) {
        this.config = config;
        this.serializationSchema = serializationSchema;
        this.key = config.get(RedisOptions.KEY);
        this.command = config.get(RedisOptions.COMMAND);
        this.primaryKeys = primaryKeys;
        this.dataType = dataType;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        redisOperator = RedisOperators.get(config);
        LOG.info("RedisSinkFunction Open");
    }

    /**
     * GenericRowData
     * +I(0,学生0,学校1,1,银河1班)
     * @param input
     * @param context
     * @throws Exception
     */
    @Override
    public void invoke(Object input, Context context) throws Exception {
        /**
         * 将SinkFunction的输入也就是insert into的select的那些列的数据强转成行格式对象RowData
         */
        RowData rowData = (RowData)input;
        /**
         * 判断当前的RowKind，如果不是insert和update_after就直接return
         */
        RowKind kind = rowData.getRowKind();
        /**
         * 这里insert就是插入，update_after就是更新。对于lpush和rpush这种命令肯定是没法支持更新了
         */
        if ((kind != RowKind.INSERT) && (kind != RowKind.UPDATE_AFTER)) {
            return;
        }
        /**
         * 将输入数据序列化，然后根据配置项中的命令参数写入redis
         */
        String value = new String(serializationSchema.serialize(rowData));
        /**
         * 执行redis命令的的key
         */
        String redisKey = key;
        /**
         * LPUSH、RPUSH、SADD三个命令直接无脑推入数据即可，只支持INSERT语义
         */
        if(RedisCmd.LPUSH.equalsIgnoreCase(command)) {
            redisOperator.lpush(redisKey,value);
        } else if(RedisCmd.RPUSH.equalsIgnoreCase(command)) {
            redisOperator.rpush(redisKey,value);
        } else if(RedisCmd.SADD.equalsIgnoreCase(command)) {
            redisOperator.sadd(redisKey,value);
        } else if(RedisCmd.SET.equalsIgnoreCase(command)){
            /**
             * set语义需要一个key，为了支持update语义，需要拿到主键的值，这里联合主键多个字段值用
             * 下划线连接
             */
            redisKey = genKey(rowData);
            redisOperator.set(redisKey,value);
        } else if(RedisCmd.HSET.equalsIgnoreCase(command)) {
            if(!StringUtils.isEmpty(redisKey)) {
                /**
                 * hset语义需要一个key和field，很明显也要拿到主键的值，这里把外部输入的key作为hset的key，
                 * 主键值作为hset的field
                 */
                redisOperator.hset(redisKey, genKey(rowData), value);
            } else {
                redisKey = getValue(rowData,0);
                /**
                 * 如果没有key配置项，则默认将第一个字段的值作为key，第二个字段的值作为field
                 */
                redisOperator.hset(redisKey, getValue(rowData,1), value);
            }
        }
        /**
         * 判断sinkKeyTtl值是否是-1，如果是-1表示一致保存在redis，如果不为-1，则设置ttl指定秒，
         * 这个配置可以用来做实时缓存预热
         */
        int ttl = config.get(RedisOptions.SINK_KEY_TTL);
        if(ttl != -1) {
            redisOperator.expire(redisKey,ttl);
        }
    }

    @Override
    public void close() throws Exception {
        if (this.redisOperator != null) {
            this.redisOperator.close();
        }
        LOG.info("RedisSinkFunction Close");
    }

    /**
     * 根据数据行和主键，生成redis的set命令的key
     * @param rowData
     * @return
     */
    private String genKey(RowData rowData) {
        if(CollectionUtil.isNullOrEmpty(primaryKeys)) {
            throw new IllegalArgumentException("primary key cant be empty");
        }
        List<String> primaryValues = new ArrayList<>();
        for(String field:primaryKeys){
            primaryValues.add(getValue(rowData,field));
        }
        return String.join(Delimiters.UNDER_LINE,primaryValues);
    }

    /**
     * 根据字段名获取字段值
     * @param rowData 行数据
     * @param fieldName 字段名
     * @return
     */
    private String getValue(RowData rowData, String fieldName) {
        int index = ((RowType) dataType.getLogicalType()).getFieldIndex(fieldName);
        return getValue(rowData, index);
    }

    /**
     * 根据字段名的位置获取字段值
     * @param rowData
     * @param fieldIdx
     * @return
     */
    private String getValue(RowData rowData, int fieldIdx) {
        LogicalType logicalType = ((RowType) dataType.getLogicalType()).getTypeAt(fieldIdx);
        return RedisRowUtil.rowDataToString(logicalType, rowData, fieldIdx);
    }



}
